# data_fetcher.py
import threading
from concurrent.futures import ThreadPoolExecutor
from okutils import utils
import pandas as pd
import datetime
import talib


# 模拟从第三方交易所API请求数据
def request_data_from_exchange(symbol, bar):
    return utils.get_mark_price_candles(symbol, bar=bar)


def convert_timestamp(timestamp):
    timestamp = int(timestamp) / 1000
    dt_object = datetime.datetime.fromtimestamp(timestamp)
    return dt_object.strftime('%Y-%m-%d %H:%M:%S')


def collate_data(data):
    # 列名
    columns = ["timestamp", "open", "high", "low", "close", "spot_volumn", "spot_volumn_100", "usdt_volumn", "confirm"]

    # 创建 DataFrame
    df = pd.DataFrame(data['data'], columns=columns)

    df['open'] = df['open'].astype(float)
    df['close'] = df['close'].astype(float)
    df['high'] = df['high'].astype(float)
    df['low'] = df['low'].astype(float)

    # 计算EMA20
    # 计算EMA20
    df.sort_index(ascending=False, inplace=True)
    df['EMA20'] = talib.EMA(df['close'], timeperiod=20)
    df.sort_index(ascending=True, inplace=True)

    df['timestamp'] = df['timestamp'].apply(convert_timestamp)

    return df


class ExchangeRequester(threading.Thread):
    def __init__(self, data_queue, symbols, bar):
        super().__init__()
        self.data_queue = data_queue
        self.symbols = symbols
        self.bar = bar

    def run(self):
        with ThreadPoolExecutor(max_workers=10) as executor:  # 并发请求数据
            for symbol in self.symbols:
                future = executor.submit(request_data_from_exchange, symbol, self.bar)
                data = future.result()  # 获取结果
                if data:
                    self.data_queue.put({
                        'data': collate_data(data),
                        'symbol': symbol
                    })  # 将数据放入队列
        self.data_queue.put(None)  # 结束标记
